package com.css.resoft.p2pmonitor.test;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.ParseFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.thrift2.generated.TColumn;
import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
import org.apache.hadoop.hbase.thrift2.generated.TResult;
import org.apache.hadoop.hbase.thrift2.generated.TScan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import com.css.resoft.p2pmonitor.util.DateUtil;

public class ScanData {
	
	public final static String slave1host = "39.107.224.196";
	public final static String slave3host = "39.107.112.148";
	public final static int port = 9090;
	
	public static void main(String[] args) throws Exception {
		
		long startTime = System.currentTimeMillis();
		
		scanData();
		
		long endTime = System.currentTimeMillis();
		System.out.println(endTime-startTime);
		System.out.println("ok");
		
	}
	
	private static void scanData() throws Exception {
		
		String table = "MONSTAT_TODO";
		
//		String table = "J0901";
//		String orgcode = "91440300068588146D";
//		String datadate = "2018-03-11";
		
//		byte [] startRow = getRowKey(orgcode,"2018-03-20","0");
		byte [] startRow =  Bytes.toBytes( System.currentTimeMillis()-30*24*3600*1000);
//		byte [] stopRow = getRowKey(orgcode,"2018-03-08","0");
		
		TTransport transport = new TSocket(slave3host,port);
		TProtocol protocol = new TBinaryProtocol(transport);
		THBaseService.Iface client = new THBaseService.Client(protocol);
		transport.open();
		TScan scan = new TScan();
//		scan.setStartRow(startRow);
//		scan.setStopRow(stopRow);
		
//		scan.setColumns(columns);
//		RowFilter rf = new RowFilter(CompareOp.EQUAL,new SubstringComparator("2018-03-11"));
//		System.out.println(rf.toString());
//		String filter = "(ColumnPrefixFilter('DDATE') AND ValueFilter(=,'substring:2018'))"
//				+ "OR ColumnPrefixFilter('XH') OR ColumnPrefixFilter('ORGCODE') ";
		String filter = "(ColumnPrefixFilter('ISDONE') AND ValueFilter(=,'substring:NO'))"
				+ "OR ColumnPrefixFilter('DATE_STRINGS') ";
		
		scan.setFilterString(filter.getBytes());
		
		List<TResult> results = client.getScannerResults(ByteBuffer.wrap(table.getBytes()), scan,10);
		
		for (TResult tResult : results) {
			
			for (TColumnValue columnValue : tResult.getColumnValues()) {
				
				System.out.println(Bytes.toString(columnValue.getQualifier())+" = " +Bytes.toString(columnValue.getValue()));
			}
			System.out.println();
		}
		
		System.out.println(results.size());
		
		transport.close();
		
	}

	private static byte[] getRowKey(String orgCode, String dataDate, String xh) {
		
		byte [] buffer = new byte[12];
		
		int days = DateUtil.convert2Day(dataDate, DateUtil.DATE_FORMAT);
		int invert_days = Integer.MAX_VALUE - days;
		
		Bytes.putInt(buffer, 0, orgCode.hashCode());
		Bytes.putInt(buffer, 4, invert_days); 
		Bytes.putInt(buffer, 8, Integer.parseInt(xh)); 
		return buffer ;
	}
	
}
